package cn.gwm.flink.streaming.sink.hbase;


import cn.gwm.utils.ConfigLoader;
import cn.hutool.json.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.List;

/**
 * @Author: liangguang
 * @Date: 2022/10/12
 * @Description:
 */
@Slf4j
public class HbaseSinkFunction extends RichSinkFunction<HbaseBean> {
    private Integer maxSize = 300;
    private Long delayTime = 1000L;

    public HbaseSinkFunction() {
    }

    public HbaseSinkFunction(Integer maxSize, Long delayTime) {
        this.maxSize = maxSize;
        this.delayTime = delayTime;
    }

    private Connection connection;
    private Long lastInvokeTime;
    private List<Put> puts = new ArrayList<>(maxSize);

    public static Connection getConnection(String zkQuorum, int port) throws Exception {
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", zkQuorum);
        conf.set("hbase.zookeeper.property.clientPort", port + "");
        Connection connection = ConnectionFactory.createConnection(conf);
        return connection;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
//        //创建一个Hbase的连接
        connection = getConnection(ConfigLoader.get("zookeeper.quorum"), 2181);
        // 获取系统当前时间
        lastInvokeTime = System.currentTimeMillis();
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }

    @Override
    public void invoke(HbaseBean bean, Context context) throws Exception {
        byte[] rowkeybyte = Bytes.toBytes(bean.getRowKey());
        Put put = new Put(rowkeybyte);
        // 添加put对象到list集合
        puts.add(put);
        long currentTime = System.currentTimeMillis();
        JSONObject jsonObj = bean.getDatamap();
        if (jsonObj != null) {
            jsonObj.forEach((k, v) -> {
                put.addColumn(Bytes.toBytes(bean.getColumnFamily()), Bytes.toBytes(k), Bytes.toBytes(String.valueOf(v)));
            });
        }

        if (puts.size() == maxSize || currentTime - lastInvokeTime >= delayTime) {
            Table table = connection.getTable(TableName.valueOf(bean.getTableName()));
            table.put(puts);
            puts.clear();
            lastInvokeTime = currentTime;
            table.close();
        }
    }


//    private static org.apache.hadoop.conf.Configuration configuration = HBaseConfiguration.create();
//    private static Connection connection = null;
//    private static BufferedMutator mutator;
//
//    @Override
//    public void open(Configuration parameters) throws Exception {
//        configuration.set("hbase.zookeeper.quorum", ConfigLoader.get("zookeeper.quorum"));
//        configuration.set("hbase.zookeeper.property.clientPort", "2181");
//        configuration.setInt("hbase.client.operation.timeout", 30000);
//        configuration.setInt("hbase.client.scanner.timeout.period", 200000);
//        connection = ConnectionFactory.createConnection(configuration);
//    }
//
//
//    @Override
//    public void invoke(HbaseBean value, Context context) throws Exception {
//        BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(value.getTableName()));
//        params.writeBufferSize(512 * 1024);
//        mutator = connection.getBufferedMutator(params);
//        Put put = new Put(Bytes.toBytes(value.getRowKey()));
//        JSONObject kv = value.getDatamap();
//        kv.forEach((k, v) -> {
//            put.addColumn(Bytes.toBytes(value.getColumnFamily()), Bytes.toBytes(k), Bytes.toBytes(String.valueOf(v)));
//        });
//        mutator.mutate(put);
//        mutator.flush();
//    }
//
//    @Override
//    public void close() throws Exception {
//        super.close();
//    }
}
